package com.yinuo.starter.factory;

import com.yinuo.starter.bean.TransactionalBean;
import com.yinuo.starter.dao.TransactionalDao;
import com.yinuo.starter.properties.JmsProperties;
import com.yinuo.starter.properties.ProducerProperties;
import com.yinuo.starter.result.ResultBase;
import com.yinuo.starter.util.Str;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.aspectj.lang.ProceedingJoinPoint;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;

import java.sql.Timestamp;
import java.util.Date;
import java.util.concurrent.*;

@Slf4j
public class ProducerFactory {

    private TransactionMQProducer producer = null;

    @Autowired
    private PlatformTransactionManager platformTransactionManager;

    @Autowired
    private TransactionDefinition transactionDefinition;

    @Autowired
    private TransactionalDao transactionalDao;

    private final String TRANS_THREAD_NAME = "yinuo-transactional-thread";

    private final Integer CORE_POOL_SIZE = 2;
    private final Integer MAXIMUM_POOL_SIZE = 5;
    private final Integer KEEP_ALIVE_TIME = 100;
    private final Integer BLOCKING_QUEUE_SIZE = 2000;

    private final Integer PRODUCER_TYPE = 1;

    public ProducerFactory(JmsProperties jmsProperties, ProducerProperties producerProperties) {
        if(producer != null){
            return;
        }
        TransactionListener transactionListener = new TransactionListenerImpl();
        producer = new TransactionMQProducer(producerProperties.getGroupName());
        producer.setNamesrvAddr(Str.checkAddr(jmsProperties.getNameSrvAddr()));
        producer.setMaxMessageSize(producerProperties.getMaxMessageSize());
        producer.setTransactionListener(transactionListener);
        producer.setExecutorService(new ThreadPoolExecutor(CORE_POOL_SIZE, MAXIMUM_POOL_SIZE, KEEP_ALIVE_TIME, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(BLOCKING_QUEUE_SIZE), r -> {
                    Thread thread = new Thread(r);
                    thread.setName(TRANS_THREAD_NAME);
                    return thread;
                }));
        producer.setSendMsgTimeout(producerProperties.getSendMsgTimeout());
        producer.setRetryTimesWhenSendAsyncFailed(producerProperties.getRetryTimesWhenSendAsyncFailed());
        producer.setRetryAnotherBrokerWhenNotStoreOK(producerProperties.isRetryAnotherBrokerWhenNotStoreOk());
        start();
        log.info("生产者启动成功");
    }

    public TransactionMQProducer getProducer() {
        return this.producer;
    }

    private void start() {
        try {
            this.producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    class TransactionListenerImpl implements TransactionListener {

        @Override
        public LocalTransactionState executeLocalTransaction(Message message, Object o) {
            TransactionStatus transactionStatus = platformTransactionManager.getTransaction(transactionDefinition);
            TransactionalBean transactionalBean = new TransactionalBean();
            transactionalBean.setId("P" + message.getKeys())
                    .setTopicName(message.getTopic())
                    .setTagName(message.getTags())
                    .setCreateTime(new Timestamp(new Date().getTime()))
                    .setModifyTime(transactionalBean.getCreateTime())
                    .setType(PRODUCER_TYPE)
                    .setKeys(message.getKeys());
            try {
                ((ProceedingJoinPoint)o).proceed();
                transactionalBean.setValue(ResultBase.TRANSLATION_COMMIT);
                transactionalDao.insert(transactionalBean);
            } catch (Throwable throwable) {
                throwable.printStackTrace();
                log.error("事务消息已被回滚");
                platformTransactionManager.rollback(transactionStatus);
                return LocalTransactionState.ROLLBACK_MESSAGE;
            }
            platformTransactionManager.commit(transactionStatus);
            log.info("事务消息已提交");
            return LocalTransactionState.COMMIT_MESSAGE;
        }

        @Override
        public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
            TransactionalBean transactionalBean = transactionalDao.selectByPrimary("P" + messageExt.getKeys(), PRODUCER_TYPE);
            if(transactionalBean != null && transactionalBean.getValue().equals(ResultBase.TRANSLATION_COMMIT)){
                log.info("事务消息已提交");
                return LocalTransactionState.COMMIT_MESSAGE;
            }else{
                log.error("事务消息已被回滚");
                return LocalTransactionState.ROLLBACK_MESSAGE;
            }
        }
    }
}
